package com.wande.dataplatform.filecollection.service.impl;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.wande.common.core.utils.StringUtils;
import com.wande.dataplatform.common.utils.DorisConnectionManager;
import com.wande.dataplatform.filecollection.common.constant.FileCollectionConstants;
import com.wande.dataplatform.filecollection.common.enums.ErrorHandlingStrategy;
import com.wande.dataplatform.filecollection.common.enums.FileCollectionErrorCode;
import com.wande.dataplatform.filecollection.common.exception.FileCollectionException;
import com.wande.dataplatform.filecollection.config.FileCollectionProperties;
import com.wande.dataplatform.filecollection.domain.CollectionRule;
import com.wande.dataplatform.filecollection.domain.MappingRule;
import com.wande.dataplatform.filecollection.domain.dto.FieldMapping;
import com.wande.dataplatform.filecollection.domain.dto.LoadResult;
import com.wande.dataplatform.filecollection.domain.dto.ValidationResult;
import com.wande.dataplatform.filecollection.domain.dto.ValidationRuleConfig;
import com.wande.dataplatform.filecollection.service.IDataLoadService;
import com.wande.dataplatform.filecollection.transform.ITransformFunction;
import com.wande.dataplatform.filecollection.transform.TransformFunctionRegistry;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.sql.*;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.regex.Pattern;

/**
 * 数据加载服务实现
 *
 * @author wande
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class DataLoadServiceImpl implements IDataLoadService {

    private final DorisConnectionManager dorisConnectionManager;
    private final TransformFunctionRegistry transformFunctionRegistry;
    private final FileCollectionProperties properties;
    private final com.wande.dataplatform.filecollection.validation.ValidationService validationService;

    @Override
    public LoadResult loadData(List<Map<String, Object>> data, CollectionRule rule, MappingRule mappingRule) {
        return loadDataBatch(data.iterator(), rule, mappingRule);
    }

    @Override
    public LoadResult loadDataBatch(Iterator<Map<String, Object>> dataIterator, CollectionRule rule, MappingRule mappingRule) {
        long startTime = System.currentTimeMillis();
        LoadResult result = LoadResult.builder()
            .success(true)
            .totalRecords(0L)
            .insertedRecords(0L)
            .updatedRecords(0L)
            .skippedRecords(0L)
            .failedRecords(0L)
            .errors(new ArrayList<>())
            .build();

        Connection connection = null;
        try {
            // 获取数据库连接
            connection = dorisConnectionManager.getConnection();
            connection.setAutoCommit(false);

            // 解析字段映射
            List<FieldMapping> fieldMappings = parseFieldMappings(mappingRule.getFieldMappings());
            
            // 根据入库策略执行不同的加载逻辑
            String loadStrategy = rule.getLoadStrategy();
            if (FileCollectionConstants.LOAD_STRATEGY_OVERWRITE.equals(loadStrategy)) {
                // 全量覆盖：先清空表
                truncateTable(connection, rule.getTargetTable());
            }

            // 批量插入数据
            int batchSize = properties.getLoader().getBatchSize();
            List<Map<String, Object>> batch = new ArrayList<>(batchSize);
            int recordIndex = 0;

            while (dataIterator.hasNext()) {
                Map<String, Object> sourceData = dataIterator.next();
                recordIndex++;
                result.setTotalRecords(result.getTotalRecords() + 1);

                try {
                    // 校验数据 - 使用新的校验服务
                    ValidationResult validationResult = validationService.validateRecord(sourceData, mappingRule, recordIndex);
                    if (!validationResult.getValid()) {
                        handleValidationError(result, recordIndex, validationResult, sourceData);
                        continue;
                    }

                    // 转换数据
                    Map<String, Object> targetData = transformData(sourceData, mappingRule);
                    batch.add(targetData);

                    // 达到批次大小，执行批量插入
                    if (batch.size() >= batchSize) {
                        executeBatch(connection, rule.getTargetTable(), fieldMappings, batch, loadStrategy, result);
                        batch.clear();
                    }
                } catch (Exception e) {
                    log.error("处理记录失败: recordIndex={}", recordIndex, e);
                    result.setFailedRecords(result.getFailedRecords() + 1);
                    result.addError(LoadResult.LoadError.builder()
                        .recordIndex(recordIndex)
                        .errorType("PROCESS_ERROR")
                        .errorMessage(e.getMessage())
                        .recordData(JSONUtil.toJsonStr(sourceData))
                        .build());
                }
            }

            // 处理剩余的批次
            if (!batch.isEmpty()) {
                executeBatch(connection, rule.getTargetTable(), fieldMappings, batch, loadStrategy, result);
            }

            // 提交事务
            connection.commit();
            
        } catch (Exception e) {
            log.error("数据加载失败", e);
            result.setSuccess(false);
            result.setErrorMessage(e.getMessage());
            
            // 回滚事务
            if (connection != null) {
                try {
                    connection.rollback();
                } catch (SQLException ex) {
                    log.error("事务回滚失败", ex);
                }
            }
        } finally {
            // 关闭连接
            if (connection != null) {
                try {
                    connection.setAutoCommit(true);
                    connection.close();
                } catch (SQLException e) {
                    log.error("关闭连接失败", e);
                }
            }
        }

        result.setElapsedTime(System.currentTimeMillis() - startTime);
        return result;
    }

    @Override
    public ValidationResult validateData(Map<String, Object> record, MappingRule mappingRule) {
        ValidationResult result = ValidationResult.success();

        if (StringUtils.isEmpty(mappingRule.getValidationRules())) {
            return result;
        }

        try {
            List<ValidationRuleConfig> validationRules = JSONUtil.toList(
                mappingRule.getValidationRules(), ValidationRuleConfig.class);

            for (ValidationRuleConfig rule : validationRules) {
                String fieldName = rule.getFieldName();
                Object fieldValue = record.get(fieldName);

                // 非空校验
                if ("NOT_NULL".equals(rule.getValidationType())) {
                    if (fieldValue == null || StringUtils.isEmpty(fieldValue.toString())) {
                        result.addError(ValidationResult.ValidationError.builder()
                            .fieldName(fieldName)
                            .errorType("NOT_NULL")
                            .errorMessage(StrUtil.isNotEmpty(rule.getErrorMessage()) ? 
                                rule.getErrorMessage() : "字段不能为空")
                            .fieldValue(fieldValue)
                            .build());
                    }
                }
                // 格式校验
                else if ("FORMAT".equals(rule.getValidationType())) {
                    if (fieldValue != null && StringUtils.isNotEmpty(fieldValue.toString())) {
                        Map<String, Object> params = JSONUtil.toBean(rule.getValidationParams(), Map.class);
                        String pattern = (String) params.get("pattern");
                        if (!Pattern.matches(pattern, fieldValue.toString())) {
                            result.addError(ValidationResult.ValidationError.builder()
                                .fieldName(fieldName)
                                .errorType("FORMAT")
                                .errorMessage(StrUtil.isNotEmpty(rule.getErrorMessage()) ? 
                                    rule.getErrorMessage() : "字段格式不正确")
                                .fieldValue(fieldValue)
                                .build());
                        }
                    }
                }
                // 范围校验
                else if ("RANGE".equals(rule.getValidationType())) {
                    if (fieldValue != null) {
                        Map<String, Object> params = JSONUtil.toBean(rule.getValidationParams(), Map.class);
                        validateRange(fieldValue, params, fieldName, result, rule.getErrorMessage());
                    }
                }
                // 枚举值校验
                else if ("ENUM".equals(rule.getValidationType())) {
                    if (fieldValue != null) {
                        Map<String, Object> params = JSONUtil.toBean(rule.getValidationParams(), Map.class);
                        List<String> values = (List<String>) params.get("values");
                        if (!values.contains(fieldValue.toString())) {
                            result.addError(ValidationResult.ValidationError.builder()
                                .fieldName(fieldName)
                                .errorType("ENUM")
                                .errorMessage(StrUtil.isNotEmpty(rule.getErrorMessage()) ? 
                                    rule.getErrorMessage() : "字段值不在允许的范围内")
                                .fieldValue(fieldValue)
                                .build());
                        }
                    }
                }
            }
        } catch (Exception e) {
            log.error("数据校验失败", e);
            result.addError(ValidationResult.ValidationError.builder()
                .fieldName("UNKNOWN")
                .errorType("VALIDATION_ERROR")
                .errorMessage("校验规则解析失败: " + e.getMessage())
                .build());
        }

        return result;
    }

    @Override
    public Map<String, Object> transformData(Map<String, Object> sourceData, MappingRule mappingRule) {
        Map<String, Object> targetData = new HashMap<>();

        try {
            List<FieldMapping> fieldMappings = parseFieldMappings(mappingRule.getFieldMappings());

            for (FieldMapping mapping : fieldMappings) {
                String sourceField = mapping.getSourceField();
                String targetField = mapping.getTargetField();
                Object sourceValue = sourceData.get(sourceField);

                // 处理空值
                if (sourceValue == null || StringUtils.isEmpty(sourceValue.toString())) {
                    if (StringUtils.isNotEmpty(mapping.getDefaultValue())) {
                        targetData.put(targetField, mapping.getDefaultValue());
                    } else if (Boolean.TRUE.equals(mapping.getNullable())) {
                        targetData.put(targetField, null);
                    }
                    continue;
                }

                // 应用转换函数
                Object transformedValue = sourceValue;
                if (StringUtils.isNotEmpty(mapping.getTransformFunction())) {
                    ITransformFunction function = transformFunctionRegistry.getFunction(mapping.getTransformFunction());
                    if (function != null) {
                        transformedValue = function.transform(sourceValue, mapping.getTransformParams());
                    }
                }

                // 类型转换
                Object finalValue = convertType(transformedValue, mapping.getTargetType());
                targetData.put(targetField, finalValue);
            }
        } catch (Exception e) {
            log.error("数据转换失败", e);
            throw new FileCollectionException(FileCollectionErrorCode.DATA_TRANSFORM_ERROR, 
                "数据转换失败: " + e.getMessage());
        }

        return targetData;
    }

    /**
     * 执行批量插入
     */
    private void executeBatch(Connection connection, String tableName, List<FieldMapping> fieldMappings,
                             List<Map<String, Object>> batch, String loadStrategy, LoadResult result) throws SQLException {
        
        if (batch.isEmpty()) {
            return;
        }

        // 构建SQL语句
        StringBuilder sql = new StringBuilder();
        List<String> targetFields = new ArrayList<>();
        for (FieldMapping mapping : fieldMappings) {
            targetFields.add(mapping.getTargetField());
        }

        if (FileCollectionConstants.LOAD_STRATEGY_UPSERT.equals(loadStrategy)) {
            // UPSERT模式：使用INSERT ... ON DUPLICATE KEY UPDATE
            sql.append("INSERT INTO ").append(tableName).append(" (");
            sql.append(String.join(", ", targetFields));
            sql.append(") VALUES (");
            sql.append(String.join(", ", Collections.nCopies(targetFields.size(), "?")));
            sql.append(") ON DUPLICATE KEY UPDATE ");
            
            List<String> updateClauses = new ArrayList<>();
            for (String field : targetFields) {
                updateClauses.add(field + " = VALUES(" + field + ")");
            }
            sql.append(String.join(", ", updateClauses));
        } else {
            // APPEND模式：普通INSERT
            sql.append("INSERT INTO ").append(tableName).append(" (");
            sql.append(String.join(", ", targetFields));
            sql.append(") VALUES (");
            sql.append(String.join(", ", Collections.nCopies(targetFields.size(), "?")));
            sql.append(")");
        }

        try (PreparedStatement pstmt = connection.prepareStatement(sql.toString())) {
            for (Map<String, Object> record : batch) {
                int paramIndex = 1;
                for (String field : targetFields) {
                    pstmt.setObject(paramIndex++, record.get(field));
                }
                pstmt.addBatch();
            }

            int[] updateCounts = pstmt.executeBatch();
            
            // 统计结果
            for (int count : updateCounts) {
                if (count > 0) {
                    if (FileCollectionConstants.LOAD_STRATEGY_UPSERT.equals(loadStrategy)) {
                        if (count == 1) {
                            result.setInsertedRecords(result.getInsertedRecords() + 1);
                        } else {
                            result.setUpdatedRecords(result.getUpdatedRecords() + 1);
                        }
                    } else {
                        result.setInsertedRecords(result.getInsertedRecords() + 1);
                    }
                } else {
                    result.setSkippedRecords(result.getSkippedRecords() + 1);
                }
            }
        } catch (SQLException e) {
            log.error("批量插入失败: sql={}", sql, e);
            result.setFailedRecords(result.getFailedRecords() + batch.size());
            result.addError(LoadResult.LoadError.builder()
                .errorType("SQL_ERROR")
                .errorMessage(e.getMessage())
                .sqlStatement(sql.toString())
                .build());
            throw e;
        }
    }

    /**
     * 清空表
     */
    private void truncateTable(Connection connection, String tableName) throws SQLException {
        String sql = "TRUNCATE TABLE " + tableName;
        try (Statement stmt = connection.createStatement()) {
            stmt.execute(sql);
            log.info("清空表成功: {}", tableName);
        }
    }

    /**
     * 解析字段映射
     */
    private List<FieldMapping> parseFieldMappings(String fieldMappingsJson) {
        if (StringUtils.isEmpty(fieldMappingsJson)) {
            return new ArrayList<>();
        }
        return JSONUtil.toList(fieldMappingsJson, FieldMapping.class);
    }

    /**
     * 范围校验
     */
    private void validateRange(Object fieldValue, Map<String, Object> params, String fieldName,
                               ValidationResult result, String errorMessage) {
        try {
            if (fieldValue instanceof Number) {
                double value = ((Number) fieldValue).doubleValue();
                if (params.containsKey("min")) {
                    double min = Double.parseDouble(params.get("min").toString());
                    if (value < min) {
                        result.addError(ValidationResult.ValidationError.builder()
                            .fieldName(fieldName)
                            .errorType("RANGE")
                            .errorMessage(StrUtil.isNotEmpty(errorMessage) ? 
                                errorMessage : "字段值小于最小值: " + min)
                            .fieldValue(fieldValue)
                            .build());
                    }
                }
                if (params.containsKey("max")) {
                    double max = Double.parseDouble(params.get("max").toString());
                    if (value > max) {
                        result.addError(ValidationResult.ValidationError.builder()
                            .fieldName(fieldName)
                            .errorType("RANGE")
                            .errorMessage(StrUtil.isNotEmpty(errorMessage) ? 
                                errorMessage : "字段值大于最大值: " + max)
                            .fieldValue(fieldValue)
                            .build());
                    }
                }
            }
        } catch (Exception e) {
            log.error("范围校验失败", e);
        }
    }

    /**
     * 处理校验错误
     */
    private void handleValidationError(LoadResult result, int recordIndex, 
                                      ValidationResult validationResult, Map<String, Object> sourceData) {
        handleValidationError(result, recordIndex, validationResult, sourceData, ErrorHandlingStrategy.SKIP_RECORD);
    }

    /**
     * 处理校验错误（支持错误处理策略）
     */
    private void handleValidationError(LoadResult result, int recordIndex, 
                                      ValidationResult validationResult, Map<String, Object> sourceData,
                                      ErrorHandlingStrategy strategy) {
        result.setFailedRecords(result.getFailedRecords() + 1);
        
        StringBuilder errorMsg = new StringBuilder();
        for (ValidationResult.ValidationError error : validationResult.getErrors()) {
            errorMsg.append(error.getFieldName()).append(": ").append(error.getErrorMessage()).append("; ");
        }
        
        // 记录错误
        result.addError(LoadResult.LoadError.builder()
            .recordIndex(recordIndex)
            .errorType("VALIDATION_ERROR")
            .errorMessage(errorMsg.toString())
            .recordData(JSONUtil.toJsonStr(sourceData))
            .build());

        // 根据策略处理
        switch (strategy) {
            case STOP_TASK:
                // 终止任务
                throw new FileCollectionException(FileCollectionErrorCode.VALIDATION_FAILED,
                    String.format("记录 %d 校验失败，终止任务: %s", recordIndex, errorMsg));
            case MARK_ERROR:
                // 标记错误，记录日志
                log.warn("记录 {} 校验失败: {}", recordIndex, errorMsg);
                break;
            case SKIP_RECORD:
            default:
                // 跳过记录，继续处理（默认行为）
                log.debug("跳过校验失败的记录 {}: {}", recordIndex, errorMsg);
                break;
        }
    }

    /**
     * 类型转换
     */
    private Object convertType(Object value, String targetType) {
        if (value == null) {
            return null;
        }

        try {
            switch (targetType.toUpperCase()) {
                case "STRING":
                case "VARCHAR":
                case "TEXT":
                    return value.toString();
                    
                case "INT":
                case "INTEGER":
                    if (value instanceof Number) {
                        return ((Number) value).intValue();
                    }
                    return Integer.parseInt(value.toString());
                    
                case "LONG":
                case "BIGINT":
                    if (value instanceof Number) {
                        return ((Number) value).longValue();
                    }
                    return Long.parseLong(value.toString());
                    
                case "DOUBLE":
                case "DECIMAL":
                    if (value instanceof Number) {
                        return ((Number) value).doubleValue();
                    }
                    return Double.parseDouble(value.toString());
                    
                case "FLOAT":
                    if (value instanceof Number) {
                        return ((Number) value).floatValue();
                    }
                    return Float.parseFloat(value.toString());
                    
                case "BOOLEAN":
                    if (value instanceof Boolean) {
                        return value;
                    }
                    return Boolean.parseBoolean(value.toString());
                    
                case "DATE":
                    if (value instanceof LocalDate) {
                        return value;
                    }
                    return LocalDate.parse(value.toString(), DateTimeFormatter.ISO_LOCAL_DATE);
                    
                case "DATETIME":
                case "TIMESTAMP":
                    if (value instanceof LocalDateTime) {
                        return value;
                    }
                    return LocalDateTime.parse(value.toString(), DateTimeFormatter.ISO_LOCAL_DATE_TIME);
                    
                default:
                    return value;
            }
        } catch (Exception e) {
            log.warn("类型转换失败: value={}, targetType={}", value, targetType, e);
            return value;
        }
    }
}
